package sms.connect.manager.smpp;

import com.zx.sms.BaseMessage;
import com.zx.sms.codec.smpp.Address;
import com.zx.sms.codec.smpp.msg.DeliverSm;
import com.zx.sms.codec.smpp.msg.DeliverSmReceipt;
import com.zx.sms.codec.smpp.msg.DeliverSmResp;
import com.zx.sms.codec.smpp.msg.SubmitSm;
import com.zx.sms.common.util.ChannelUtil;
import com.zx.sms.connect.manager.EndpointEntity;
import com.zx.sms.connect.manager.EventLoopGroupFactory;
import com.zx.sms.connect.manager.ExitUnlimitCirclePolicy;
import com.zx.sms.connect.manager.ServerEndpoint;
import com.zx.sms.session.cmpp.SessionState;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;
import org.apache.commons.lang3.RandomUtils;
import org.marre.sms.SmsAlphabet;
import org.marre.sms.SmsDcs;
import org.marre.sms.SmsMsgClass;
import org.marre.sms.SmsTextMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sms.handler.api.gate.SessionConnectedHandler;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

public class SMPPSessionConnectedHandler extends SessionConnectedHandler {

	private static final Logger logger = LoggerFactory.getLogger(SMPPSessionConnectedHandler.class);

	private static	String content="你好中文短信";
	private static String sourceAddress="13800138000";
	private static String destAddress="10086";
	private static Thread thr;


	public SMPPSessionConnectedHandler(Thread thr, int t,String content,String sourceAddress,String destAddress) {
		this.thr = thr;
		totleCnt = new AtomicInteger(t);
		if (content !=null){
			this.content = content;
		}
		if (sourceAddress !=null){
			this.sourceAddress = sourceAddress;
		}
		if (destAddress !=null){
			this.destAddress = destAddress;
		}
	}

	@Override
	protected BaseMessage createTestReq(String str) {
		final EndpointEntity finalentity = getEndpointEntity();
		if (finalentity instanceof ServerEndpoint) {
			DeliverSm pdu = new DeliverSm();
	        pdu.setSourceAddress(new Address((byte)0,(byte)0,sourceAddress));
	        pdu.setDestAddress(new Address((byte)0,(byte)0,destAddress));
	        pdu.setSmsMsg(new SmsTextMessage(content,SmsDcs.getGeneralDataCodingDcs(SmsAlphabet.GSM,SmsMsgClass.CLASS_UNKNOWN)));
			return pdu;
		} else {
			SubmitSm pdu = new SubmitSm();
			pdu.setRegisteredDelivery((byte)0);
	        pdu.setSourceAddress(new Address((byte)0,(byte)0,destAddress));
	        pdu.setDestAddress(new Address((byte)0,(byte)0,sourceAddress));
	        pdu.setSmsMsg(new SmsTextMessage(content,SmsDcs.getGeneralDataCodingDcs(SmsAlphabet.UCS2,SmsMsgClass.CLASS_UNKNOWN)));
			return pdu;
		}
	}
	
	public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {

		if (msg instanceof DeliverSmReceipt) {
			DeliverSmReceipt e = (DeliverSmReceipt) msg;
			
			DeliverSmResp res =e.createResponse();
			res.setMessageId(String.valueOf(System.currentTimeMillis()));
			 ctx.writeAndFlush(res);

		} else {
			ctx.fireChannelRead(msg);
		}
	}

	@Override
	public void userEventTriggered(final ChannelHandlerContext ctx, Object evt) throws Exception {
		final AtomicInteger tmptotal = totleCnt;
		if (evt == SessionState.Connect) {

			final Channel ch = ctx.channel();
			EventLoopGroupFactory.INS.submitUnlimitCircleTask(new Callable<Boolean>() {

				@Override
				public Boolean call() throws Exception {
					int cnt = RandomUtils.nextInt() & 0x4fff;
					Promise<BaseMessage> frefuture = null;
					while (cnt > 0 && tmptotal.get() > 0) {
						List<Promise<BaseMessage>> futures = null;
						ChannelFuture chfuture = null;
						BaseMessage msg = createTestReq(UUID.randomUUID().toString());
						chfuture = ChannelUtil.asyncWriteToEntity(getEndpointEntity().getId(), msg);
//						futures = ChannelUtil.syncWriteLongMsgToEntity(getEndpointEntity().getId(), msg);

//						chfuture = ctx.writeAndFlush(msg);

						if (chfuture != null)
							chfuture.sync();
						cnt--;
						tmptotal.decrementAndGet();
						if (futures != null) {
							try {
								for (Promise<BaseMessage> future : futures) {
									future.addListener(new GenericFutureListener<Future<BaseMessage>>() {
										@Override
										public void operationComplete(Future<BaseMessage> future) throws Exception {
											if (future.isSuccess()) {
//												 logger.info("response:{}",future.get());
											} else {
//												 logger.error("response:{}",future.cause());
											}
										}
									});
									frefuture = future;
								}

							} catch (Exception e) {
								e.printStackTrace();
								cnt--;
								tmptotal.decrementAndGet();
								break;
							}
						}else {
							//连接不可写了，等待上一个response回来
							//再把消息发出去
							ctx.writeAndFlush(msg);

							if(frefuture!=null) {
								frefuture.sync();
								frefuture = null;
							}else {
								break;
							}
						}
					}
					return true;
				}
			}, new ExitUnlimitCirclePolicy<Boolean>() {
				@Override
				public boolean notOver(Future<Boolean> future) {
					if (future.cause() != null)
						future.cause().printStackTrace();

					boolean over = ch.isActive() && tmptotal.get() > 0;
					if (!over) {
						logger.info("========send over.============");
						LockSupport.unpark(thr);
						// ch.writeAndFlush(new CmppTerminateRequestMessage());
					}
					return over;
				}
			}, 1);

		}
		ctx.fireUserEventTriggered(evt);
	}

}
